package com.nepu.gmall.realtime.selfLearming;

import com.alibaba.fastjson.JSON;
import com.nepu.gmall.realtime.util.DateFormatUtil;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.common.functions.ReduceFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.configuration.Configuration;
import org.apache.flink.configuration.RestOptions;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.KeyedStream;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.streaming.api.functions.windowing.WindowFunction;
import org.apache.flink.streaming.api.windowing.assigners.TumblingProcessingTimeWindows;
import org.apache.flink.streaming.api.windowing.time.Time;
import org.apache.flink.streaming.api.windowing.windows.TimeWindow;
import org.apache.flink.util.Collector;

/**
 * @author chenshuaijun
 * @create 2023-07-26 16:02
 */
public class MiniBatchProcess {
    public static void main(String[] args) throws Exception {
        Configuration conf = new Configuration();
        conf.set(RestOptions.ENABLE_FLAMEGRAPH, true);
        StreamExecutionEnvironment env = StreamExecutionEnvironment.createLocalEnvironmentWithWebUI(conf);


        env.setParallelism(3);

        DataStreamSource<String> orderInfoDataStream = env.addSource(new GenerateSource());


        SingleOutputStreamOperator<Tuple2<String, Integer>> mapDataStream = orderInfoDataStream.map(new MapFunction<String, Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> map(String value) throws Exception {
                return Tuple2.of(value, 1);
            }
        });

        KeyedStream<Tuple2<String, Integer>, String> keyedStream = mapDataStream.keyBy(value -> value.f0);

        SingleOutputStreamOperator<String> reduce = keyedStream.window(TumblingProcessingTimeWindows.of(Time.seconds(10))).reduce(new ReduceFunction<Tuple2<String, Integer>>() {
            @Override
            public Tuple2<String, Integer> reduce(Tuple2<String, Integer> value1, Tuple2<String, Integer> value2) throws Exception {
                return Tuple2.of(value1.f0, value1.f1 + value2.f1);
            }
        }, new WindowFunction<Tuple2<String, Integer>, String, String, TimeWindow>() {
            @Override
            public void apply(String s, TimeWindow window, Iterable<Tuple2<String, Integer>> input, Collector<String> out) throws Exception {
                Tuple2<String, Integer> next = input.iterator().next();
                OrderInfoBean testOrderInfo = new OrderInfoBean();
                testOrderInfo.setStt(DateFormatUtil.toYmdHms(window.getStart()));
                testOrderInfo.setEdt(DateFormatUtil.toYmdHms(window.getEnd()));
                testOrderInfo.setProvinceId(next.f0);
                testOrderInfo.setTraderNum((long) next.f1);
                out.collect(JSON.toJSONString(testOrderInfo));
            }
        });

        reduce.print("testOrder");


        env.execute();

    }
}
